package com.asiainfo.dacp.datastash.broker.leader.service;

import java.util.Date;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.StashBrokerBoot;
import com.asiainfo.dacp.datastash.broker.core.BrokerContext;
import com.asiainfo.dacp.datastash.broker.core.vo.Broker;
import com.asiainfo.dacp.datastash.broker.util.ZKUtil;
import com.asiainfo.dacp.datastash.orm.dao.core.DataStashBrokerDao;
import com.asiainfo.dacp.datastash.orm.po.core.DataStashBrokerPo;
import com.asiainfo.dacp.util.JsonHelper;
import com.google.common.base.Preconditions; 
 

@Service
public class MonitorBrokerService {
	private static Logger LOG = LoggerFactory.getLogger(MonitorBrokerService.class);
	@Autowired
	private DataStashBrokerDao brokerDao;
	
	public void service(){
		LOG.info("monitor broker [[start]]..");
		try {
			ZKUtil.watchPathChild(BrokerContext.ACTIVE_BROKERS, new PathChildrenCacheListener() {
				@Override
				public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
					//path: /datastash/broker/active/0
					//data: {"brokerId":"0","brokerIp":"10.37.129.2","maxNet":null} 
					
					switch (event.getType()) {
					case CHILD_ADDED: 
					case CHILD_UPDATED:
					case CHILD_REMOVED:  
						addOrUpdate(event);
						break;
					default:
						break;
					}
				}
			});
		} catch (Exception e) { 
			LOG.error("monitor broker [[error]]..", e);
			e.printStackTrace();
		}
		LOG.info("monitor broker [[end]]..");
	}
	
	
	public void addOrUpdate(PathChildrenCacheEvent event){
		DataStashBrokerPo po = buildPo(event);
		brokerDao.save(po);
		
	}
	
	public DataStashBrokerPo buildPo(PathChildrenCacheEvent event){
		String brokerData= new String(event.getData().getData());
		
		Broker broker = JsonHelper.getInstance().read(brokerData, Broker.class);
		
		DataStashBrokerPo po = new DataStashBrokerPo();
		
		Preconditions.checkNotNull(broker.getBrokerId(), "zk信息中broker id不能为空，请检查 .");
		
		po.setBrokerId(broker.getBrokerId());
		po.setBrokerIp(broker.getBrokerIp());
		
		switch (event.getType()) {
		case CHILD_ADDED: 
		case CHILD_UPDATED: 
			po.setIsAlive("1");
			break;
		case CHILD_REMOVED:  
			po.setIsAlive("0");
			break;
		default:
			break;
		} 
		
		if(broker.getBrokerId().equals(BrokerContext.getInstance().getBrokerId())){
			po.setIsLeader("1");
		}else{
			po.setIsLeader("0");
		}
		
		po.setStatDate(new Date());
		
		return po;
	}
}
